package com.atguigu.union;

import com.atguigu.bean.WaterSensor;
import com.atguigu.function.WaterSensorFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author yhm
 * @create 2024-04-03 9:35
 */
public class Test02_Connect {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 2. 读取数据源
        DataStreamSource<String> dataStreamSource1 = env.socketTextStream("hadoop102", 7777);
        DataStreamSource<String> dataStreamSource2 = env.socketTextStream("hadoop102", 8888);
        SingleOutputStreamOperator<WaterSensor> dataStream3 = dataStreamSource2.flatMap(new WaterSensorFunction());

        // 3. 处理数据
        // connect允许连接的两个流数据类型不同
        // 调用connect方法的流称作主流 -> 泛型第一个
        // 传入的参数是连接流  -> 泛型是第二个
        ConnectedStreams<String, WaterSensor> connectedStreams = dataStreamSource1.connect(dataStream3);

        // 连接流方法和一般数据流类似  也包含map flatmap process
        // 填写的函数不同   带有Co开头
        // 具有三个泛型  <主流类型,连接流类型,输出数据类型>
        // 会有两个抽象方法等你写
        // 编写完成之后 返回结果又变成一个数据类型的基础流
        SingleOutputStreamOperator<Integer> processStream = connectedStreams.process(new CoProcessFunction<String, WaterSensor, Integer>() {


            /**
             * 处理主流
             * @param value  主流的数据
             * @param ctx   环境上下文
             * @param out   输出工具
             * @throws Exception
             */
            @Override
            public void processElement1(String value, Context ctx, Collector<Integer> out) throws Exception {
                // 输出vc
                // 主流vc + 10
                System.out.println("处理主流数据");
                String[] split = value.split(",");
                if (split.length == 3) {
                    out.collect(Integer.parseInt(split[2]) + 10);
                }
            }

            /**
             *  处理连接流
             * @param value  连接流的数据
             * @param ctx   环境上下文
             * @param out   输出工具
             * @throws Exception
             */
            @Override
            public void processElement2(WaterSensor value, Context ctx, Collector<Integer> out) throws Exception {
                // 输出vc
                // 连接流vc + 20
                System.out.println("处理连接流数据");
                out.collect(value.vc + 20);
            }
        });

        // 4. 输出
        processStream.print();

        // 5. 执行环境
        env.execute();
    }
}
